/* 
 * This file part of cget - A libcurl based downloading tool.
 * http://lianwei3.googlepages.com/home2
 * Copyright (C) 2006 Wei Lian <lianwei3@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Library General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include "downloader.h"
#include "header.h"
#include "mycurses.h"
#include "util.h"
int pipe_sig[2];
pthread_mutex_t lock;
MyCurses *mycurses;

void handle_sig_int(int signo)
{
    if (signo == SIGINT)
	write(pipe_sig[1], "1", 1);
}


Downloader::Downloader()
{
    signal(SIGINT, handle_sig_int);
    pipe(pipe_sig);
    pthread_mutex_init(&lock, NULL);
    ///init nmycurses
    mycurses = new MyCurses;
}


Downloader::~Downloader()
{
    delete mycurses;
}


size_t Downloader::my_write_func(char *ptr, size_t size, size_t nmemb,
				 struct ThreadInfo *info)
{
    bool cancel;
    pthread_mutex_lock(&lock);
    fseeko(info->fp, info->seek, SEEK_SET);	//share
    size_t nsize = size * nmemb;
    size_t dis = 0;
    while (nsize) {
	size_t re = fwrite(ptr + dis, 1, nsize, info->fp);	//share
	nsize -= re;
	dis += re;
    }
    info->seek += dis;		//dis;
    cancel = info->cancel;
    pthread_mutex_unlock(&lock);
    if (cancel)
	return 0;
    return dis;			//dis;
}

int Downloader::my_progress_func(struct ThreadInfo *info,
				 double t, double d, double ut, double un)
{
    char str[mycurses->width()];
    snprintf(str, sizeof(str),	//        "%ld,%ld",info->seek,info->end);
	     "(%d) download=%10.f, totoal=%10.f, (%3.f %%)%s", info->id,
	     d, t, d * 100.0 / t, info->url);
    mycurses->mvprint(info->id, str);
    return 0;
}

void *Downloader::thread_download(void *stream)
{
    struct ThreadInfo *info = (struct ThreadInfo *) stream;
    std::vector < ThreadInfo > *vec_info = info->vecptr;
    ///common args
    if (strncmp(info->url, "ftp", 3) == 0) {
	curl_easy_setopt(info->curl,
			 CURLOPT_FTP_FILEMETHOD, CURLFTPMETHOD_NOCWD);
    }
    curl_easy_setopt(info->curl, CURLOPT_URL, info->url);
    curl_easy_setopt(info->curl, CURLOPT_FOLLOWLOCATION, TRUE);///redirect
    curl_easy_setopt(info->curl, CURLOPT_WRITEFUNCTION, my_write_func);
    curl_easy_setopt(info->curl, CURLOPT_WRITEDATA, info);
    curl_easy_setopt(info->curl, CURLOPT_NOPROGRESS, FALSE);
    curl_easy_setopt(info->curl,
		     CURLOPT_PROGRESSFUNCTION, my_progress_func);
    curl_easy_setopt(info->curl, CURLOPT_PROGRESSDATA, info);
    curl_easy_setopt(info->curl, CURLOPT_USERAGENT,
		     "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.0)");
    ///loop
    off_t __seek, __end;
    while (1) {
	///set range
	pthread_mutex_lock(&lock);
	__seek = info->seek;
	__end = info->end;
	info->cancel = FALSE;
	pthread_mutex_unlock(&lock);
	assert(__seek >= 0);
	if (__seek < __end + 1) {	///job unfinished
	    char range[64];
	    sprintf(range, "%ld-%ld", __seek, __end);
	    curl_easy_setopt(info->curl, CURLOPT_RANGE, range);
	    ///execute
	    CURLcode res = curl_easy_perform(info->curl);
	    ///check downloading result
	    pthread_mutex_lock(&lock);
	    __seek = info->seek;
	    __end = info->end;
	    pthread_mutex_unlock(&lock);
	    if (__seek < __end + 1) {	///job unfinished
		if (res != CURLE_OK && res != CURLE_WRITE_ERROR) {	///error occured
		    pthread_mutex_lock(&lock);
		    mycurses->mvprint(vec_info->size() + info->id,
				      info->id, res, info->seek, info->end,
				      info->length);
		    pthread_mutex_unlock(&lock);
		    sleep(3);
		}
		continue;
	    }
	}
	///reallocate job between the slowest and the finished thread
	pthread_mutex_lock(&lock);
	int slowest_id = reallocate(vec_info, info->working_thread_first,
				    info->working_thread_last, info->id,
				    102400);
	if (slowest_id != -1) {
	    assert(slowest_id != info->id);
	    vec_info->at(slowest_id).cancel = TRUE;	///restart the slowest thread
	}
	pthread_mutex_unlock(&lock);
	if (slowest_id == -1)
	    break;
    }
    ///curl cleanup
    curl_easy_cleanup(info->curl);
    ///write quit signal
    write(info->pipe[1], "1", 1);
    return NULL;
}

void Downloader::execute_download(std::vector
				  < struct ThreadInfo >*vec_info,
				  uint num_task, char *outfile_cg)
{
    int error;
    FILE *fp = vec_info->at(0).fp;

    for (uint j = 0; j < vec_info->size(); j += num_task) {

	uint min = min_value(vec_info->size(), j + num_task);
	///create pthread
	for (uint i = j; i < min; i++) {
	    vec_info->at(i).working_thread_first = j;
	    vec_info->at(i).working_thread_last = min;
	    ///
	    error =
		pthread_create(vec_info->at(i).thread, NULL,
			       thread_download, &(vec_info->at(i)));
	    assert(error == 0);
	}
	///loop
	std::set < uint > entry;
	char tmp[3];
	///
	while (1) {
	    fd_set rfds;
	    FD_ZERO(&rfds);
	    int maxfd = -1;
	    for (uint i = j; i < min; i++) {
		FD_SET(vec_info->at(i).pipe[0], &rfds);
		maxfd = max_value(maxfd, vec_info->at(i).pipe[0]);
	    }
	    FD_SET(pipe_sig[0], &rfds);
	    maxfd = max_value(maxfd, pipe_sig[0]);
	    ///select
	    int retval = select(maxfd + 1, &rfds, NULL, NULL, NULL);
	    if (retval > 0) {
		////is all task finished?
		for (uint i = j; i < min; i++) {
		    if (FD_ISSET(vec_info->at(i).pipe[0], &rfds)) {
			read(vec_info->at(i).pipe[0], tmp, sizeof(tmp));
			entry.insert(i - j);
		    }
		}
		if (entry.size() == min - j)
		    break;
		////signal int catched
		if (FD_ISSET(pipe_sig[0], &rfds)) {
		    read(pipe_sig[0], tmp, sizeof(tmp));
		    ///cancel download
		    pthread_mutex_lock(&lock);
		    for (uint i = j; i < min; i++)
			pthread_cancel(*(vec_info->at(i).thread));
		    pthread_mutex_unlock(&lock);
		    for (uint i = j; i < min; i++) {
			error =
			    pthread_join(*(vec_info->at(i).thread), NULL);
			assert(error == 0);
		    }
		    ////truncate to original length
		    truncate(outfile_cg, vec_info->at(0).length);
		    ///write info for later download resume
		    int num = 0;
		    for (uint i = 0; i < vec_info->size(); i++)
			if (vec_info->at(i).seek <= vec_info->at(i).end)
			    num++;
		    fseeko(fp, vec_info->at(0).length, SEEK_SET);
		    fwrite(&num, sizeof(uint), 1, fp);
		    for (uint i = 0; i < vec_info->size(); i++)
			if (vec_info->at(i).seek <= vec_info->at(i).end) {
			    fwrite(&vec_info->at(i).seek,
				   sizeof(off_t), 1, fp);
			    fwrite(&vec_info->at(i).end,
				   sizeof(off_t), 1, fp);
			}
		    fclose(fp);
		    delete mycurses;
		    exit(0);
		}
	    }
	}
	for (uint i = j; i < min; i++) {
	    error = pthread_join(*(vec_info->at(i).thread), NULL);
	    assert(error == 0);
	}
    }

}
